package com.example.soscall.utils;





import com.example.soscall.callback.NotifyListener;

import org.reactivestreams.Subscriber;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.FlowableTransformer;
import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by Administrator on 2018/6/13.
 */

public class RxUtils {
    private static Scheduler scheduler;
    /**
     * 延时执行
     *
     * @param time
     * @param timeUnit
     * @param listener
     */
    public static Disposable delayDo(long time, TimeUnit timeUnit, NotifyListener listener) {
        if (time < 0) time = 0;
        return Flowable
                .timer(time, timeUnit)
                .compose(io2main())
                .subscribe((aLong -> {
                    listener.onNotify(null, null);
                }), Throwable::printStackTrace);

    }

    /**
     * 每隔一段时间执行
     *
     * @param time
     * @param timeUnit
     * @param listener
     * @return
     */
    public static Disposable intervel( int time, TimeUnit timeUnit, NotifyListener listener) {
        return Flowable
                .interval(0, time, timeUnit)
                .compose(io2main())

                .subscribe((aLong -> {
                    listener.onNotify(null, null);
                }), Throwable::printStackTrace);


    }

    public static <T> FlowableTransformer<T, T> io2main() {
        if (scheduler == null) {
            scheduler = Schedulers.from(new ThreadPoolExecutor(7, 20, 20L,
                    TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(64)
                    , new ThreadPoolExecutor.DiscardOldestPolicy()));
        }
        return tObservable -> tObservable.subscribeOn(scheduler).unsubscribeOn(scheduler).observeOn(AndroidSchedulers.mainThread());
    }



}
